<?php

/**
 * Net_Conveyor
 *
 * Conveyor is an HTTP queue that is like TiVo for your data. It allows you to
 * push data into channels and then have multiple consumers consume that data.
 * Additionally consumers can rewind channels and replay them.
 *
 * PHP versions 5.2.0+
 *
 * Copyright (c) 2008, Digg, Inc.
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without 
 * modification, are permitted provided that the following conditions are met:
 *
 * - Redistributions of source code must retain the above copyright notice, 
 *   this list of conditions and the following disclaimer.
 * - Redistributions in binary form must reproduce the above copyright notice, 
 *   this list of conditions and the following disclaimer in the documentation 
 *   and/or other materials provided with the distribution.
 * - Neither the name of the Digg, Inc. nor the names of its contributors 
 *   may be used to endorse or promote products derived from this software 
 *   without specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * @category  Net
 * @package   Net_Conveyor
 * @author    Joe Stump <joe@joestump.net> 
 * @copyright 2008 Digg, Inc.
 * @license   http://tinyurl.com/new-bsd-license New BSD License
 * @version   CVS: $Id:$
 * @link      http://code.google.com/p/digg
 * @link      http://theryanking.com/entries/2008/02/26/introducing-conveyor/
 * @link      http://theryanking.com/hg/conveyor
 */

require_once 'Validate.php';
require_once 'Net/Conveyor/Exception.php';

/**
 * Net_Conveyor
 *
 * @category Net
 * @package  Net_Conveyor
 * @author   Joe Stump <joe@joestump.net>
 * @license  http://tinyurl.com/new-bsd-license New BSD License
 * @link     http://code.google.com/p/digg
 * @link     http://theryanking.com/entries/2008/02/26/introducing-conveyor/
 * @link     http://theryanking.com/hg/conveyor
 */
class Net_Conveyor
{
    /**
     * Conveyor host
     *
     * @var string $host Conveyor host URI 
     */
    protected $host = '';

    /**
     * Channel name
     *
     * @var string $channel Conveyor channel name
     */
    protected $channel = '';

    /**
     * Constructor
     *
     * @param string $host    Conveyor host URI
     * @param string $channel Name of channel to communicate with
     *
     * @throws {@link Net_Conveyor_Exception} on invalid host/channel
     * @return void
     */
    public function __construct($host, $channel)
    {
        if (!Validate::uri($host)) {
            throw new Net_Conveyor_Exception('Invalid conveyor host');
        }

        if (!self::isValidChannelName($channel)) {
            throw new Net_Conveyor_Exception('Invalid channel name');
        }

        $this->channel = $channel;
        $this->host    = $host;
    }

    /**
     * Post data into the channel
     *
     * @param mixed $data The raw data to post into the queue
     * 
     * @throws {@link Net_Conveyor_Exception} on HTTP errors
     * @return boolean True if all goes well
     */
    public function post($data)
    {
        if (is_object($data) || is_array($data)) {
            throw new Net_Conveyor_Exception('Date must not be complex');
        }

        $ch = curl_init($this->host . '/channels/' . $this->channel);
        curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'POST'); 
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_POSTFIELDS, $data);
        curl_setopt($ch, CURLOPT_HTTPHEADER, array(
            'Date: ' . date("D, d M Y H:i:s T")
        ));

        $result = self::sendRequest($ch, array(202));

        return true;
    }

    /**
     * Get a specific item from the stream
     *
     * @param integer $id The ID of the item from the stream
     *
     * @throws {@link Net_Conveyor_Exception} on HTTP errors
     * @return mixed
     */
    public function get($id)
    {
        $url = $this->host . '/channels/' . $this->channel . '/' . (int)$id;
        $ch  = curl_init($url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        return self::sendRequest($ch, array(200));
    }

    /**
     * Get the channel's status
     *
     * @throws {@link Net_Conveyor_Exception} on HTTP errors
     * @return array
     */
    public function status()
    {
        $url = $this->host . '/channels/' . $this->channel;
        $ch  = curl_init($url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        $res = self::sendRequest($ch, array(200));
        return json_decode($res, true);
    }

    /**
     * Get the next item from Conveyor
     *
     * Parameters:
     * - group  string  The Conveyor consumer group to fetch next for
     * - count  integer How many things to fetch from the queue
     * 
     * @param array $params An array of parameters
     *
     * @throws UnexpectedArgumentException on invalid parameters
     * @throws {@link Net_Conveyor_Exception} on bad server responses
     * @return mixed The next data (JSON decoded if count > 1)
     */
    public function next(array $params = array())
    {
        $url = $this->host . '/channels/' . $this->channel . '?next';

        $sets = array();
        foreach ($params as $key => $val) {
            switch ($key) {
            case 'group':
                $sets[] = $key . '=' . urlencode($val);
                break;
            case 'count':
                $sets[] = 'n=' . (int)$val;
            default:
                throw new UnexpectedArgumentException(
                    'Invalid parameter specified'
                );
            }
        }

        $url = $url . (count($sets) ? '&' . implode('&', $sets) : '');

        $ch = curl_init($url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

        $body = self::sendRequest($ch, array(200));
        if (isset($params['count'])) {
            return json_decode($body);
        }

        return $body;
    }

    /**
     * Rewinde the conveyor belt
     *
     * Parameters:
     * - id     integer A conveyor ID to rewind back to
     * - time   integer A UNIX timestamp to rewind back to
     * - group  string  The Conveyor consumer group to fetch next for
     * 
     * @param array $params An array of parameters for rewind request
     *
     * @throws UnexpectedArgumentException on malformed $params array
     * @throws {@link Net_Conveyor_Exception} on HTTP errors
     * @return mixed 
     */
    public function rewind(array $params)
    {
        $url = $this->host . '/channels/' . $this->channel;

        $sets = array();
        if (isset($params['id']) && isset($params['group'])) {
            $sets[] = 'rewind_id=' . (int)$params['id'];
            $sets[] = 'group=' . $params['group'];
        } elseif (isset($params['id'])) {
            $sets[] = 'rewind_id=' . (int)$params['id'];
        } elseif (isset($params['time']) && isset($params['group'])) {
            $sets[] = 'rewind_time=' . (int)$params['time'];
            $sets[] = 'group=' . $params['group'];
        } elseif (isset($params['time'])) {
            $sets[] = 'rewind_time=' . (int)$params['time'];
        } else {
            throw new UnexpectedArgumentException(
                'Parameters are invalid'
            );
        }

        $ch = curl_init($url . '?' . implode('&', $sets));
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'POST'); 

        $resp = self::sendRequest($ch, array(200));
        return true; 
    }

    /**
     * Execute the HTTP request
     *
     * @param resource $ch       The curl handle to execute on
     * @param array    $expected Array of expected HTTP codes
     *
     * @throws UnexpectedArgumentException on invalid arguments
     * @throws {@link Net_Conveyor_Exception} on HTTP failures
     * @return mixed The body of the HTTP request
     */
    static protected function sendRequest($ch, array $expected)
    {
        if (!is_resource($ch)) {
            throw new UnexpectedArgumentException('Invalid curl handle');
        }

        if (!count($expected)) {
            throw new UnexpectedArgumentException('Invalid expected array');
        }

        $result = curl_exec($ch);
        if ($result === false) {
            throw new Net_Conveyor_Exception(
                curl_error($ch), curl_errno($ch)
            );
        }


        $code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        if (!in_array((int)$code, $expected)) {
            throw new Net_Conveyor_Exception('Unexpected HTTP code');
        }

        curl_close($ch);

        return $result;
    }

    /**
     * Create a conveyor channel
     *
     * @param string $host    The HTTP host string for the conveyor host
     * @param string $channel The new channel name
     *
     * @throws {@link Net_Conveyor_Exception} on errors
     * @return object Instance of {@link Net_Conveyor} on success
     */
    static public function createChannel($host, $channel)
    {
        if (!Validate::uri($host)) {
            throw new Net_Conveyor_Exception('Invalid conveyor host');
        }

        if (!self::isValidChannelName($channel)) {
            throw new Net_Conveyor_Exception('Invalid channel name');
        }

        $url = $host . '/channels/' . $channel;
        $ch  = curl_init($url);
        curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'PUT'); 
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

        self::sendRequest($ch, array(201, 202));

        $instance = new Net_Conveyor($host, $channel);
        return $instance;
    }

    /**
     * Is channel name valid?
     *
     * @param string $channel The name of the channel to check
     *
     * @see Net_Conveyor::__construct()
     * @see Net_Conveyor::createChannel()
     * @return boolean
     */
    static protected function isValidChannelName($channel)
    {
        return (preg_match('/^[A-Z0-9\-]+$/i', $channel)); 
    }
}

?>
